package io.wen.bd.s6m3.spark

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.graphx.{Edge, Graph}

object AirportGraphApp {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setAppName(this.getClass.getCanonicalName).setMaster("local[*]")
    val sc = new SparkContext(conf)
    sc.setLogLevel("WARN")

    // 构建图
    val verticesRDD = sc.makeRDD(Array((1L, "SFO"), (2L, "ORD"), (3L, "DFW")))
    val edgesRDD = sc.makeRDD(Array(Edge(1L, 2L, 1800), Edge(2L, 3L, 800), Edge(3L, 1L, 400)))

    val graph = Graph(verticesRDD, edgesRDD)

    // 求所有的顶点
    println("========= vertices")
    graph.vertices.collect().foreach(println)

    // 求所有的边
    println("========= edges")
    graph.edges.collect().foreach(println)

    // 求所有的triplets
    println("========= triplets")
    graph.triplets.collect().foreach(println)

    // 求顶点数
    println("========= vertices count")
    println(graph.vertices.count())

    // 求边数
    println("========= edges count")
    println(graph.edges.count())

    // 求机场距离大于1000的有几个
    println("========= distance > 1000")
    val gt1000 = graph.triplets.filter(t => t.attr > 1000)
    gt1000.foreach(println)
    println(gt1000.count)

    // 按所有机场之间的距离排序（降序
    println("========= sort by distance")
    graph.triplets.sortBy(- _.attr).collect().foreach(println)
  }
}
